import com.study.reactor.vo.UserPO;
import com.study.reactor.vo.UserVO;
import org.junit.Test;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.*;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.*;

/**
 * @author yuantongqin
 * desc:
 * 2020-11-16
 */
public class Operater {

    @Test
    public void publishOn(){
        CountDownLatch latch = new CountDownLatch(1);
        Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);

        final Flux<String> flux = Flux.create(new Consumer<FluxSink<Integer>>() {
            @Override
            public void accept(FluxSink<Integer> fluxSink) {
                for (int i = 0; i < 3; i++) {
                    System.out.println(Thread.currentThread().getName()+"==内容"+i);
                    fluxSink.next(i);
                }

                fluxSink.complete();
            }
        })

                .map(i -> 10 + i)
                .publishOn(s)
                .map(i -> "value " + i);



        new Thread(() -> {
            System.out.println("执行了吗");
            flux.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.out.println(Thread.currentThread().getName()+"==内容"+s);
            }
        });}).start();

        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void parallel(){
        CountDownLatch latch = new CountDownLatch(1);
        Flux.create(new Consumer<FluxSink<Integer>>() {
            @Override
            public void accept(FluxSink<Integer> fluxSink) {
                for (int i = 0; i < 10; i++) {
                    System.out.println(Thread.currentThread().getName()+"==发射="+i);
                    fluxSink.next(i);
                }

                fluxSink.complete();
            }
        }).publishOn(Schedulers.elastic()).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) {
                System.out.println(Thread.currentThread().getName()+"==内容"+integer);
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void buffer(){

        Flux.just(1,2,3,4,5,6,7,8).groupBy(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) {
                return integer % 3;
            }
        }).subscribe(new Consumer<GroupedFlux<Integer, Integer>>() {
            @Override
            public void accept(GroupedFlux<Integer, Integer> integerIntegerGroupedFlux) {
                integerIntegerGroupedFlux.subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) {
                        System.out.println("分组id="+integerIntegerGroupedFlux.key()+" ==值=="+integer);
                    }
                });
            }
        });



    }

    @Test
    public void reduce(){

        Flux.just(1,2,3,4,5).reduce(new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) {
                System.out.println(integer+"==过程1");
                System.out.println(integer2+"===过程2");
                return integer+ integer2;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) {
                System.out.println(integer+"==结果");
            }
        });

    }

    @Test
    public void merge() throws InterruptedException {

        // 模仿数据库查询
        List<UserPO> userList = new ArrayList<>();
        UserPO userPO = new UserPO();
        userPO.setName("张三");
        userPO.setMobilePhone("1321111111");
        userList.add(userPO);
        // 模仿数据查询地址列表
        EmitterProcessor<List<String>> emitterProcessor = EmitterProcessor.create();
        List<String> addressList = new ArrayList<>();
        addressList.add("北京");
        addressList.add("上海");
        addressList.add("广告");
        emitterProcessor.onNext(addressList);
        // 对查询出来的数据进行逻辑处理
        Flux.fromIterable(userList).map(new Function<UserPO, UserVO>() {
            @Override
            public UserVO apply(UserPO userPO) {
                UserVO userVO = new UserVO();
                userVO.setName(userPO.getName());
                String mobilePhone = userPO.getMobilePhone();
                StringBuilder sb = new StringBuilder();
                // 电话号码隐藏处理
                sb.append(mobilePhone, 0, 3).append("***").append(mobilePhone.substring(mobilePhone.length() - 3));
                userVO.setPhone(sb.toString());
                return userVO;
            }
        }).zipWith(emitterProcessor, new BiFunction<UserVO, List<String>, UserVO>() {
            @Override
            public UserVO apply(UserVO userVO, List<String> strings) {
                userVO.setShippingAddress(strings);
                return userVO;
            }
        }).subscribe(new Consumer<UserVO>() {
            @Override
            public void accept(UserVO userVO) {
                System.out.println(userVO);
            }
        });
    }

    @Test
    public void zipWith(){
        // 模仿数据库查询
        List<UserPO> userList = new ArrayList<>();
        UserPO userPO = new UserPO();
        userPO.setName("张三");
        userPO.setMobilePhone("1321111111");
        userList.add(userPO);

        // 对查询出来的数据进行逻辑处理
        Flux.fromIterable(userList).map(new Function<UserPO, UserVO>() {
            @Override
            public UserVO apply(UserPO userPO) {
                UserVO userVO = new UserVO();
                userVO.setName(userPO.getName());
                String mobilePhone = userPO.getMobilePhone();
                StringBuilder sb = new StringBuilder();
                // 电话号码隐藏处理
                sb.append(mobilePhone, 0, 3).append("***").append(mobilePhone.substring(mobilePhone.length() - 3));
                userVO.setPhone(sb.toString());
                return userVO;
            }
        }).mergeWith(new Publisher<UserVO>() {
            @Override
            public void subscribe(Subscriber<? super UserVO> s) {
                UserVO userVO = new UserVO();
                userVO.setName("li");
                s.onNext(userVO);
            }
        }).subscribe(new Subscriber<UserVO>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(Integer.MAX_VALUE);
            }
            @Override
            public void onNext(UserVO userVO) {
                System.out.println(userVO.toString());
            }
            @Override
            public void onError(Throwable t) {
            }
            @Override
            public void onComplete() {
                System.out.println("完成");
            }
        });

    }

    @Test
    public void flatMap(){
        // 模仿数据库查询
        List<UserPO> userList = new ArrayList<>();
        UserPO userPO = new UserPO();
        userPO.setName("张三");
        userPO.setMobilePhone("1321111111");
        userList.add(userPO);
        // 对查询出来的数据进行逻辑处理
        Flux.fromIterable(userList).flatMap(new Function<UserPO, Publisher<Set<Character>>>() {
            @Override
            public Publisher<Set<Character>> apply(UserPO userPO) {
                char[] chars = userPO.getMobilePhone().toCharArray();
                Set<Character> phones = new HashSet<>();
                for (char aChar : chars) {
                    phones.add(aChar);
                }
                return Flux.just(phones);
            }
        }).subscribe(new Consumer<Set<Character>>() {
            @Override
            public void accept(Set<Character> characters) {
                System.out.println(characters.toString());
            }
        });

    }

    @Test
    public void map(){

        // 模仿数据库查询
        List<UserPO> userList = new ArrayList<>();
        UserPO userPO = new UserPO();
        userPO.setName("张三");
        userPO.setMobilePhone("1321111111");
        userList.add(userPO);
        // 对查询出来的数据进行逻辑处理
        List<UserVO> block = Flux.fromIterable(userList).map(new Function<UserPO, UserVO>() {
            @Override
            public UserVO apply(UserPO userPO) {
                UserVO userVO = new UserVO();
                userVO.setName(userPO.getName());
                String mobilePhone = userPO.getMobilePhone();
                StringBuilder sb = new StringBuilder();
                // 电话号码隐藏处理
                sb.append(mobilePhone, 0, 3).append("***").append(mobilePhone.substring(mobilePhone.length() - 3));
                userVO.setPhone(sb.toString());
                return userVO;
            }
        }).collectList().block();
        System.out.println(block.toString());
    }

    @Test
    public void time() throws InterruptedException {

        CountDownLatch latch = new CountDownLatch(1);
        Flux.just("1","2","3").delayElements(Duration.ofSeconds(2)).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.out.println(Thread.currentThread().getName()+"=="+s);
            }
        });
        latch.await();
//        Flux.create(new Consumer<FluxSink<Integer>>() {
//            @Override
//            public void accept(FluxSink<Integer> fluxSink) {
//                fluxSink.next(1);
//                try {
//                    TimeUnit.SECONDS.sleep(3);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//                fluxSink.next(3);
//                fluxSink.next(4);
//                fluxSink.next(5);
//                fluxSink.next(6);
//            }
//        }).timeout(Duration.ofSeconds(2), new Publisher<Integer>() {
//            @Override
//            public void subscribe(Subscriber<? super Integer> s) {
//                s.onNext(2);
//            }
//        }).subscribe(new Subscriber<Integer>() {
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(3);
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.out.println("内容："+integer);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//
//            }
//
//            @Override
//            public void onComplete() {
//                System.out.println("完成");
//            }
//        });

//        CountDownLatch latch = new CountDownLatch(1);
//        Flux.interval(Duration.ofSeconds(2),java.time.Duration.ofSeconds(3)).subscribe(new Consumer<Long>() {
//            @Override
//            public void accept(Long aLong) {
//                System.out.println(aLong+"=="+Thread.currentThread().getName());
//            }
//        });
//        latch.await();

    }

    @Test
    public void concat(){

        Flux.error(new Error("错误")).subscribe(new Subscriber<Object>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(2);
            }

            @Override
            public void onNext(Object o) {
                System.out.println("=="+o);
            }

            @Override
            public void onError(Throwable t) {
                System.out.println("内容："+t.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("完成");
            }
        });

//        Flux.from(new Publisher<Integer>() {
//            @Override
//            public void subscribe(Subscriber<? super Integer> s) {
//                s.onNext(1);
//                s.onNext(1);
//                s.onComplete();
//            }
//        }).subscribe(new Subscriber<Integer>() {
//            @Override
//            public void onSubscribe(Subscription s) {
//
//            }
//            @Override
//            public void onNext(Integer integer) {
//                System.out.println("消费："+integer);
//            }
//            @Override
//            public void onError(Throwable t) {
//                System.out.println("错误");
//            }
//            @Override
//            public void onComplete() {
//                System.out.println("完成");
//            }
//        });

//        Flux.fromIterable(new HashMap<String,String>()).subscribe(new Subscriber<String>() {
//            @Override
//            public void onSubscribe(Subscription s) {
////                s.request(Integer.MAX_VALUE);
//            }
//            @Override
//            public void onNext(String s) {
//                System.out.println("s:"+s);
//            }
//            @Override
//            public void onError(Throwable t) {
//                System.out.println("错误");
//            }
//            @Override
//            public void onComplete() {
//                System.out.println("完成");
//            }
//        });
    }


    @Test
    public void retry(){

        Flux.just(1,2).handle(new BiConsumer<Integer, SynchronousSink<Integer>>() {
            @Override
            public void accept(Integer integer, SynchronousSink<Integer> stringSynchronousSink) {
                if(integer == 1){
                    stringSynchronousSink.error(new RuntimeException("错误1"));
                } else {
                    stringSynchronousSink.next(integer+10);
                }
            }
        }).retry(2, new Predicate<Throwable>() {
            @Override
            public boolean test(Throwable throwable) {
                System.out.println("retry重试："+throwable.getMessage());
                if(throwable instanceof RuntimeException){
                    RuntimeException re = (RuntimeException) throwable;
                    System.out.println("错误内容:"+re.getMessage());
                    return true;
                }
                return false;
            }
        }).onErrorResume(new Function<Throwable, Publisher<Integer>>() {
            @Override
            public Publisher<Integer> apply(Throwable throwable) {
                System.out.println("用0替代");
                return Flux.just(0);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer s) {
                System.out.println("内容:"+s);
            }
        });

    }

    @Test
    public void backOff(){
        Flux.just("1","2","3").map(new Function<String, Integer>() {
            @Override
            public Integer apply(String s) {
                return Integer.valueOf(s);
            }
        }).handle(new BiConsumer<Integer, SynchronousSink<String>>() {

            @Override
            public void accept(Integer integer, SynchronousSink<String> stringSynchronousSink) {

                if(integer == 2){
                    stringSynchronousSink.error(new RuntimeException("错误"));
                }
                stringSynchronousSink.next(integer+"");
            }
        }).retryBackoff(2, Duration.ofSeconds(2),Duration.ofSeconds(10)).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.out.println("内容："+s);
            }
        });
        System.out.println("结束");
    }

    @Test
    public void aab(){
        ArrayList<Integer> integers = new ArrayList<>();
        for (int i = 0; i < 6; i++) {
            integers.add(i);
        }
        Flux.fromIterable(integers).buffer(10).subscribe(new Consumer<List<Integer>>() {
            @Override
            public void accept(List<Integer> integers) {
                System.out.println(integers.size());
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) {

            }
        }, new Runnable() {
            @Override
            public void run() {
                System.out.println("完成了吗");
            }
        });


    }
    @Test
    public void a() {

//        System.out.println(a.equals(Long.valueOf(1L)));

        ArrayList<Integer> integers = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            integers.add(i);
        }

        Map<String,Integer> map = new HashMap<>();
        map.put("key1",1);
        map.put("key2",2);
        map.put("key3",3);
        map.put("key4",4);

//        Flux.fromIterable(map.entrySet()).parallel(4).runOn(Schedulers.parallel()).map(new Function<Map.Entry<String, Integer>, String>() {
//            @Override
//            public String apply(Map.Entry<String, Integer> stringIntegerEntry) {
//                String key = stringIntegerEntry.getKey();
//                Integer value = stringIntegerEntry.getValue();
//                System.out.println(Thread.currentThread().getName()+"=="+key+"=="+value);
//                return key;
//            }
//        }).doOnNext(new Consumer<String>() {
//            @Override
//            public void accept(String s) {
//                System.out.println(s);
//            }
//        }).sequential().blockLast();

        Flux.create(new Consumer<FluxSink<Integer>>() {
            @Override
            public void accept(FluxSink<Integer> fluxSink) {
                for (int i = 0; i < 10; i++) {
                }
                fluxSink.next(2);
                fluxSink.next(10);
                fluxSink.complete();
            }
        }).parallel().runOn(Schedulers.immediate())
                .flatMap(new Function<Integer, Publisher<String>>() {
                    @Override
                    public Publisher<String> apply(Integer integer) {
                        if(integer == 2){
                            System.out.println(Thread.currentThread().getName()+"来了");
                            return Flux.just();
                        }
                        Map<String,Integer> a = new HashMap<>(50);
                        for (int i = 0; i < integer * 5; i++) {
                            a.put("key"+i,i);
                        }
                        return Flux.fromIterable(a.entrySet()).map(new Function<Map.Entry<String, Integer>, String>() {
                            @Override
                            public String apply(Map.Entry<String, Integer> stringIntegerEntry) {
                                return stringIntegerEntry.getKey();
                            }
                        });
                    }
                }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String integer) {

                System.out.println(Thread.currentThread().getName() + "中途" + integer);
            }
        }).sequential().blockLast()
              ;


    }



    @Test
    public void ab() {


//        Mono.just(1).zipWith(Mono.just(2), new BiFunction<Integer, Integer, Integer>() {
//            @Override
//            public Integer apply(Integer integer, Integer integer2) {
//                return integer+integer2;
//            }
//        }).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) {
//                System.out.println(integer);
//            }
//        });

        EmitterProcessor<Integer> a = EmitterProcessor.create();
        a.onNext(1);
        a.onComplete();

        EmitterProcessor<Integer> b = EmitterProcessor.create();
        b.onNext(1);
        b.onComplete();


        Flux.zip(a, b, new BiFunction<Integer, Integer, Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) {
                return integer + integer2;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) {
                System.out.println(Thread.currentThread().getName() + integer);
            }
        });

    }

    @Test
    public void aa() {

        List<UserVO> resultData = new ArrayList<>();
        UserVO v1 = new UserVO();
        v1.setName("张三");
        v1.setAge(10);
        resultData.add(v1);

        List<UserVO> resultData2 = new ArrayList<>();
        UserVO v2 = new UserVO();
        v2.setName("嫦娥");
        v2.setAge(18);
        resultData2.add(v2);
        List<UserVO> block = Flux.zip(Flux.just(resultData), Flux.just(resultData2), new BiFunction<List<UserVO>, List<UserVO>, List<UserVO>>() {
            @Override
            public List<UserVO> apply(List<UserVO> userVOS, List<UserVO> userVOS2) {
                userVOS.addAll(userVOS2);
                return userVOS;
            }
        }).parallel(2).runOn(Schedulers.parallel()).sequential().blockLast();

        System.out.println(block);
    }

    @Test
    public void service(){
        CountDownLatch latch = new CountDownLatch(1);
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 20; i++) {

            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+"==");
                    try {
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    @Test
    public void dup() throws InterruptedException {


        ExecutorService executorService = Executors.newFixedThreadPool(10);

        CountDownLatch latch = new CountDownLatch(1);
        Flux.create(new Consumer<FluxSink<Integer>>() {
            @Override
            public void accept(FluxSink<Integer> fluxSink) {
                for (int i = 0; i < 20; i++) {

                    fluxSink.next(i);
                }
                fluxSink.complete();
            }
        }).flatMap(new Function<Integer, Publisher<String>>() {
            @Override
            public Publisher<String> apply(Integer integer) {
                List<String> a = new ArrayList<>(integer);
                for (int i = 0; i < 1; i++) {
                    a.add(i + "*" + integer);
                }


                return Flux.fromIterable(a);
            }
        }).filter(new Predicate<String>() {
            @Override
            public boolean test(String s) {
                if (s.equals("1*1")) {
                    return false;
                }
                return true;
            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String integer) {
                executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        System.out.println(Thread.currentThread().getName() + "==" + integer);
                        try {
                            TimeUnit.SECONDS.sleep(2);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        }).blockLast();

        System.out.println("完成了吗");
//        latch.await();
    }


    @Test
    public void handler() {
        UserVO userVO= new UserVO();
        userVO.setName("张安");
        Flux.just(userVO).doOnNext(new Consumer<UserVO>() {
            @Override
            public void accept(UserVO s) {
                // 改变值
                s.setName("李四");
            }
        }).subscribe(new Consumer<UserVO>() {
            @Override
            public void accept(UserVO integer) {
                System.out.println("内容="+integer);
            }
        });
    }


    @Test
    public void generate() {
        Flux.generate(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return null;
            }
        }, new BiFunction<Integer, SynchronousSink<Integer>, Integer>() {

            @Override
            public Integer apply(Integer integer, SynchronousSink<Integer> integerSynchronousSink) {
                return null;
            }
        }).limitRequest(10).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) {

            }
        });
        Flux.generate(new Consumer<SynchronousSink<String>>() {

            @Override
            public void accept(SynchronousSink<String> stringSynchronousSink) {
                // next方法只能被调用一次
                stringSynchronousSink.next("发射1");
                stringSynchronousSink.next("发射11");
                stringSynchronousSink.next("发射111");
                stringSynchronousSink.complete();
            }
        }).limitRequest(10).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.out.println(s);
            }
        });
    }

    private Subscription sub;

    public void zipwith() {

        Flux.create(new Consumer<FluxSink<Integer>>() {
            @Override
            public void accept(FluxSink<Integer> fluxSink) {
                while (fluxSink.requestedFromDownstream() == 0) {
                    // 表示管道中没有数据事件
                }
                fluxSink.onRequest(new LongConsumer() {
                    @Override
                    public void accept(long value) {
                        fluxSink.next((int) value);
                    }
                });
            }
        }).flatMap(new Function<Integer, Publisher<Tuple2<Integer, String>>>() {

            @Override
            public Publisher<Tuple2<Integer, String>> apply(Integer integer) {
                Flux.just(integer).zipWith(new Publisher<String>() {

                    @Override
                    public void subscribe(Subscriber<? super String> s) {

                    }
                }, new BiFunction<Integer, String, Tuple2<Integer, String>>() {
                    @Override
                    public Tuple2<Integer, String> apply(Integer integer, String s) {
                        return null;
                    }
                });
                return null;
            }
        }).subscribe(new Subscriber<Tuple2<Integer, String>>() {
            @Override
            public void onSubscribe(Subscription s) {
                sub = s;
            }

            @Override
            public void onNext(Tuple2<Integer, String> objects) {

            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }
        });


        Mono<List<Integer>> listMono = Flux.create(new Consumer<FluxSink<Integer>>() {
            @Override
            public void accept(FluxSink<Integer> fluxSink) {

            }
        }).flatMap(new Function<Integer, Publisher<Integer>>() {
            @Override
            public Publisher<Integer> apply(Integer integer) {

                Mono<Integer> just = Mono.just(1);
                Mono<Integer> just1 = Mono.just(2);
                Mono<Integer> integerMono = just.zipWith(just1, new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) {
                        return integer + integer2;
                    }
                });

                return integerMono;
            }
            // 收集
        }).collectList();
        // 获取到值
        List<Integer> block = listMono.block();

    }


    /**
     * 00ms内没有发出任何光，则传播错误。
     * 如果发生错误，请退回到cacheService。
     * 链的其余部分与前面的示例相似。
     */
    public void timeout() {


        Flux
                .create(new Consumer<FluxSink<String>>() {
                    @Override
                    public void accept(FluxSink<String> fluxSink) {
                        try {
                            // 延迟一秒发送事件
                            TimeUnit.SECONDS.sleep(1);
                            fluxSink.next("");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                })

                .timeout(Duration.ofMillis(800))
                .onErrorResume(new Function<Throwable, Publisher<? extends String>>() {
                    @Override
                    public Publisher<? extends String> apply(Throwable throwable) {
                        return null;
                    }
                }).flatMap(new Function<String, Publisher<Integer>>() {

            @Override
            public Publisher<Integer> apply(String s) {
                return null;
            }
        }).switchIfEmpty(new Publisher<Integer>() {
            @Override
            public void subscribe(Subscriber<? super Integer> s) {

            }
        }).take(5)
                // 我们要处理UI线程中的每个数据。
                .publishOn(Schedulers.immediate()).subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {

            }

            @Override
            public void onNext(Integer integer) {

            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }
        });

    }
}
